package source;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author itcast
 * Date 2021/6/16 11:29
 * 需求: 拆分数据
 * 开发步骤:
 * //1.env
 * //2.Source 比如 1-20之间的数字
 * //定义两个输出tag 一个奇数 一个偶数,指定类型为Long
 * //对source的数据进行process处理区分奇偶数
 * //3.获取两个侧输出流
 * //4.sink打印输出
 * //5.execute
 */
public class SplitDataStream {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source 比如 1-20之间的数字
        DataStreamSource<Long> source = env.fromSequence(1, 21);
        //定义两个侧输出流 一个奇数 一个偶数, 指定类型为Long
        OutputTag<Long> odd = new OutputTag<>("odd", TypeInformation.of(Long.class));
        OutputTag<Long> even = new OutputTag<>("even", TypeInformation.of(Long.class));
        //对source的数据进行process处理区分奇偶数
        SingleOutputStreamOperator<Long> result = source.process(new ProcessFunction<Long, Long>() {
            @Override
            public void processElement(Long value, Context context, Collector<Long> out) throws Exception {
                if (value % 2 == 0) {
                    context.output(even, value);
                } else {
                    context.output(odd, value);
                }

            }
        });
        //3.获取两个侧输出流
        //result.print();

        result.getSideOutput(even).print("偶数");
        result.getSideOutput(odd).print("奇数");
        //4.sink打印输出
        //5.execute
        env.execute();
    }
}
